package gbench.sandbox.jdbc;

import org.junit.jupiter.api.Test;

import gbench.commonApp.data.DataApp;
import gbench.common.matlib.MatlibCanvas.NVec;
import gbench.common.matlib.MatlibCanvas.Seq;
import gbench.common.matlib.MatlibCanvas.XVec;
import gbench.common.matlib.data.DataReader.DFrame;
import gbench.common.matlib.matrix.Complex;
import gbench.common.matlib.matrix.IComplex;
import gbench.common.tree.LittleTree.CronTime;
import gbench.common.tree.LittleTree.IRecord;
import gbench.common.tree.LittleTree.Jdbc;
import gbench.common.tree.LittleTree.SQL;

import static gbench.common.tree.LittleTree.IRecord.*;
import static gbench.common.matlib.MatlibCanvas.println;
import static gbench.common.tree.LittleTree.*;
import static gbench.common.tree.LittleTree.SQL.*;

import java.sql.*;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.concurrent.RecursiveTask;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Collector;
import java.util.stream.Stream;

public class JunitDB {
    
    // 数据操作类
    final class MyDataApp extends DataApp {
        public MyDataApp(String dbname) { this.reload(this.configureDbName(dbname));}
        public IRecord configureDbName(final String dbname) { return 
            cfg.derive("url", MFT("jdbc:mysql://localhost:3306/{0}?serverTimezone=GMT%2B8", dbname));
        }
    }; // MyDataApp 数据数据库操作类
    
    @Test
    public void foo() {
        final var dataApp = new MyDataApp("db1");
        final var person = IRecord.builder("name,height,weight");
        final var ptable = person.get(String.class,Long.class,Double.class);
        final var createsqls = SQL.of("person", ptable).createTable();
        
        //
        dataApp.withTransaction(sess->{
            TUPLES(createsqls).filter(e->e._1()!=1).map(e->e._2()).forEach(trycatch2(sess::sqlexecute));// 数据表重新建立 
            //
            sess.sqlexecute(nsql("insert into person values ${ foreach u in users (u.name, u.height, u.weight) }",
                REC("users",L(
                    REC("name", "张三", "height", 170, "weight",60),
                    REC("name", "李四", "height", 189, "weight",70)
            ))));
            //
            sess.sql2records(nsql("select * from person",REC())).forEach(System.out::println);
        });
    }
    
    @Test
    public void salepro() {
        final var dataApp = new MyDataApp("salepro");
        dataApp.withTransaction(sess->{
          final var rr = sess.sql2records(
              nsql("select goods_name,clientele_name,total from `order` where year(create_at)=#year limit ##size",
              REC("year",2019,"size",45)));
          rr.forEach(e->{
              System.out.println(e);
          });
      });
    }
    
    /**
     * 数据库连接 查看
     */
    @Test
    public void db1_limit() {
        final var dataApp = new MyDataApp("db1");
        // 采用 短路法获取数据流
        final var stream = dataApp.jdbcApply(j->j.sql2recordS("select * from t_gdp")).limit(3).map(e->e); // 短路流
        stream.forEach(e->{
            println(e);
        });
        println("此时数据库连接没有关闭,需要进行手动关闭, 一下三种方法任意一种:\nstream.close()，dataApp.jdbcAccept(j->j.clear(stream)),dataApp.close()");
        
        // 关闭数据库连接 的方法 一下三个方法 任意使用一种
        // stream.close(); // 关闭数据库连接
        // dataApp.jdbcAccept(j->j.clear(stream)); // 关闭数据库连接
        dataApp.close(); // 关闭数据库连接,请在此 设置断点进行 数据连接 进行查看。
        
        println("此时数据库连接，已经关闭了，可以 打开 mysql workbench: 运行 show processlist; 进行查看");
    }
    
    /**
     * 数据库连接 查看
     */
    @Test
    public void db2_limit() {
        final var dataApp = new MyDataApp("db1");
        
        // 对于 withTransaction 系列方法, jdbc 会在 withTransaction 结束时候 自动 清空 其中的 为关闭的 数据流
        dataApp.withTransaction(sess->{
            final var dfm = sess.sql2recordS("select * from t_gdp").limit(3).collect(DFrame.dfmclc); // 短路流
            sess.clear();
            println(dfm);
        }); // withTransaction
        
        println("此时数据库连接，已经关闭了，可以 打开 mysql workbench: 运行 show processlist; 进行查看");
    }
    
    /**
     * jdbc 连接数据库
     */
    @Test
    public void db3_jdbc() {
        final var dataApp = new DataApp() {}.reload2( // 加载数据库配置
                REC("url", "jdbc:mysql://localhost:3306/db1?useSSL=false&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=True"));  // dataApp
        
        dataApp.withConnection(conn -> { // 完全 使用 数据库 连接 进行 数据 操作
            final var stmt = conn.createStatement();
            final var rs = stmt.executeQuery("select * from t_gdp limit 10");
            
            final var n = rs.getMetaData().getColumnCount(); // 获取列数量
            final var indices = NVec.of(n, i -> i + 1).ints(); // 列号索引
            final var rb = Jdbc.linebuilder(rs, indices); // 行记录生成器
            
            rs.next(); // 移动光标到第一行记录
            Stream.iterate( // 需要注意 iterate 是 执行 时序是 {seed, [hasNext, action, next]*} 的 结构，于是 最后一次next 生成的 数据 不会被输出到流。
                Jdbc.readlineA(rs, indices), // seed
                p -> p != null, // hasNext
                p -> trycatch4(ResultSet::next).test(rs) ? Jdbc.readlineA(rs, indices) : null // next
            ).map(rb::build).forEach(e -> { // action
                println(e);
            }); // forEach

            println("rs", rs.isClosed());
            println("stmt", stmt.isClosed());
            println("conn", conn.isClosed());

            rs.close();
            stmt.close();
            return "finished";
        }); // withConnection 会 自动关闭 Connection
        
        println("运行结束");
    }
    
    /**
     * jdbc 连接数据库
     * 
     * GDP 数据 <br>
     * 数据制备，可参照:gbench.sandbox.data.JunitGdp 的 程序 <br>
     * 
     * id	rankid	nation	continent	gdp	gdpvalue	percent	year	createtime <br>
     * 1	1	美国	美洲	21.43万亿 (21,427,700,000,000)	21427700000000	0.244186	2019	2020-08-27 11:45:58 <br>
     * 2	2	中国	亚洲	14.34万亿 (14,342,902,842,915)	14342902842915	0.163449	2019	2020-08-27 11:45:58 <br>
     * 3	3	日本	亚洲	5.08万亿 (5,081,769,542,379)	5081769542379	0.057911	2019	2020-08-27 11:45:58 <br>
     * 4	4	德国	欧洲	3.85万亿 (3,845,630,030,823)	3845630030823	0.043824	2019	2020-08-27 11:45:58 <br>
     * 5	5	印度	亚洲	2.88万亿 (2,875,142,314,811)	2875142314811	0.032765	2019	2020-08-27 11:45:58 <br>
     * 
     */
    @Test
    public void db4_jdbc() {
        final var dataApp = new DataApp() {}.reload2( // 加载数据库配置
                REC("url", "jdbc:mysql://localhost:3306/db1?useSSL=false&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=True"));  // dataApp
        
        dataApp.withConnection(conn -> { // 完全 使用 数据库 连接 进行 数据 操作
            final var stmt = conn.createStatement(); // 生成查询语句
            final var rs = stmt.executeQuery("select * from t_gdp where year = 2019"); // 生成结果集
            
            final var dfm = Jdbc.readlineS(rs).collect(DFrame.dfmclc); // 数据查询
            final var total = dfm.dbln("gdpvalue").sum(); // gdpvalue 列求和
            final var cumsum_ratios = dfm.dbln("gdpvalue").accum() // 逐个词累加
                    .fmap(e -> NVec.of(e)).fmap(e -> e.sum() / total); // 生成占比（相对于列和)
            final var individual_ratios = cumsum_ratios.slidingS(2, 1, true) // 切分首尾相连的二元序列
                    .map(xvec -> NVec.of(xvec).classInterval()) // 计算二元差值
                    .collect(NVec.nveclc()) // 规约成 NVec 向量
                    .insert(0, cumsum_ratios.get(0)); // 复制cumsum_ratios的首行元素, 构造 独立占比向量
            
            println(dfm.head(5));
            println("gdp 的 核算统计\n", 
                dfm.cbind( // 列绑定
                    "累积占比", cumsum_ratios, // 累积占比
                    "独立占比", individual_ratios // 独立占比
                ).dfrm(r -> r.filter("id,nation,累积占比,独立占比".split(","))));
            
            // 数据库连接状态查看
            println("rs", rs.isClosed());
            println("stmt", stmt.isClosed());
            println("conn", conn.isClosed());

            rs.close(); // 关闭
            stmt.close(); // 关闭语句
            
            return "finished";
        }); // withConnection 会 自动关闭 Connection

        println("运行结束");
    }

    @Test
    public void db5_jdbc() {
        // println(NVec.of(100,i->Math.random()).collect(FVec.fveclc()));
        final var dataApp = new DataApp() {}
        .reload2("url","jdbc:mysql://localhost:3306/db1?useSSL=false&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=True");
        dataApp.withTransactionX(DFrame.dfmclc, sess -> {
            final var dfm = sess.sql2u("select * from t_gdp limit 5");
            println(dfm);
            println(sess.psql2u("select * from t_gdp where id between ? and ?",1,3));
            println(sess.psql2u(SQL.nsql("select * from t_gdp where id between #start and ?", REC("start",5)),REC("1",6)));
        });
    }
    
    @Test
    public void db6_jdbc() {
        final var url = "jdbc:mysql://localhost:3306/erp?useSSL=false&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=True";
        final var dataApp = new DataApp() {}.reload2(REC("url", url)); // dataApp
        final var result = dataApp.withTransaction(sess -> {
            // sql2u 会把 collector 的 结果 保存到 sess 的 result属性中, 并且 withTransaction 会 记录 最后一个 sql2u 的 result 结果
            sess.sql2u("select * from t_fortune where headquarter = '美国'", Collector.of(
                (Supplier<List<IRecord>>) ArrayList::new, List::add,
                (left, right) -> { left.addAll(right); return left; },
                rr -> IRecord.ROWS2COLS(rr).mutate(DFrame::new)
                    .filter3("name,revenue").rfilter((i, r) -> i > 3 && i < 20)
           )); // sql2u
        }); // dfm 位于 result 字段之中.
        println(result);
        //println(result.findOne(DFrame.class));
    }

    @Test
    public void bar() {
        
        System.out.println(
            nsql("update t_student set ${ foreach fld in kvset %fld.name=fld.value } where code=##code",
                REC("kvset",L(
                    REC("name", "id", "value", 437),
                    REC("name", "name", "value", "张三"),
                    REC("name", "sex", "value", "男"),
                    REC("name", "address", "value", "上海市徐汇区")),
                    "code",234)
                ).string2()
        );
        
        System.out.println(
            nsql("select * from t_student where id in ( ${ foreach id in users id } )",
            REC("users",L(1,2,3,4,5))).string2()
        );
        
        System.out.println(
            nsql("insert into user values ${ foreach u in users (u.name, u.height, u.weight) }",
            REC("users",L(
                REC("name", "张三", "height", 170, "weight",60),
                REC("name", "李四", "height", 189, "weight",70)
            ))).string2()
        );
    }
    
    /**
     * 求和
     * @param stream
     * @param mapper
     * @return
     */
    public Double sum(Stream<?> stream, Function<IRecord, Double> mapper) {
        return stream.map(e -> (IRecord) e).collect(Collectors.summarizingDouble(e -> mapper.apply(e))).getSum();
    }
    
    /**
     * 求和
     */
    @Test
    public void qux() {
        final var dataApp = new MyDataApp("db1");
        final var acc = "银行存款,营业收入,库存商品,营业成本".split(",");
        final var n = acc.length;
        final var rnd = new Random();
        dataApp.withTransaction(sess->{
            final var proto = REC("#id",1,"name","","drcr",1,"amount",1.0);
            TUPLES(SQL.of("entry",proto).createTable(true)).filter(t->t._1()!=1).map(t->t._2())
            .forEach(trycatch2(sess::sqlexecute));
            
            final var sb = SQL.builder("entry");
            final var rb = IRecord.builder(proto.filter2("#id"));
            NATS(200).map(i->sb.get(rb.get(acc[rnd.nextInt(n)],i%2,Math.random()*1000)).insert())
            .forEach(trycatch2(sess::sqlexecute));
            
            sess.sql2records("select * from entry").stream().collect(pvtclc("name,drcr"))
            .dfs_eval_forall((path,ones)->{
                System.out.println(path+"--->"+sum(ones,e->e.dbl("amount")));
                return ones;
            });
        });
    }
    
    /**
     * 数值数据插入:RecursiveTask 的 ForkJoin线程池
     */
    @Test
    final void num() {
        final var dataApp = new MyDataApp("db1");
        final var jdbc =  dataApp.getDataBase().getProxy().findOne(Jdbc.class);
        jdbc.withTransaction(sess -> {
            final var proto = REC("#id",1,"num",1,"createtime",new Date());
            TUPLES(SQL.of("t_num",proto).createTable(true)).filter(t->t._1()!=1).map(t->t._2())
            .forEach(trycatch2(sess::sqlexecute));
        });

        @SuppressWarnings("serial")
        class MyTask extends RecursiveTask<IComplex> {
            // 构造函数
            public MyTask(final Integer... data) {
                this.data = data;
            }
            
            // 任务分解
            @Override
            protected IComplex compute() {
                if (this.data.length > 1000) { // 任务分解
                    return ((Seq<Integer>) () -> this.data).halveS(compose_f(MyTask::fork,MyTask::new)) // 任务分解均分
                        //.parallel()
                        .map(e->e.join()).reduce(IComplex::add).get(); // 任务计算
                } else { // 批量插入
                    jdbc.withTransaction( sess -> sess.sqlexecute(
                        nsql("insert into t_num(num,createtime) values $ { FOREACH e IN recs (e.num, e.createtime) }"
                        , REC("recs", XVec.OF(this.data).fmap(e->REC("num",e,"createtime",CronTime.sdf.format(new Date()))).list())
                    ))); // withTransaction
                    println(LocalDateTime.now(),this.data);
                    return Complex.C(NVec.of(data).sum());
                }// if
            }// compute

            private Integer[] data;// 任务数据
        }; // MyTask

        println("mytask:\n",new MyTask(NVec.of(10000, i -> i + 1).intA()).invoke());
        final var sql = "select sum(num) total from t_num"; // 统计函数
        final var dfm = jdbc.withConnection(conn -> Jdbc.readlineS(conn.prepareStatement(sql).executeQuery()).collect(DFrame.dfmclc));
        println("dfm:\n",dfm.rowsL().get(0));
    }

}
